package com.pjmike.client.netty;

import com.pjmike.common.protocol.RpcDecoder;
import com.pjmike.common.protocol.RpcEncoder;
import com.pjmike.common.protocol.RpcRequest;
import com.pjmike.common.protocol.RpcResponse;
import com.pjmike.common.protocol.serialize.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PreDestroy;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @description:
 * @author: pjmike
 * @create: 2019/03/29 10:58
 *
 * netty中的channel和jdk nio包下的channel是什么关系？
 *     netty是一个基于jdk nio 之上的一个封装框架，所有功能还是基于jdk nio包的基础来完成的
 *
 *
 *
 */
@Slf4j
public class NettyClient {

    /**
     * 创建 Bootstrap。
     * 指定 EventLoopGroup 用来监听事件。
     * 定义 Channel 的传输模式为 NIO（Non-BlockingInputOutput）。
     * 设置服务器的 InetSocketAddress。
     * 在创建 Channel 时，向 ChannelPipeline 中添加一个 EchoClientHandler 实例。
     * 连接到远程节点，阻塞等待直到连接完成。
     * 阻塞，直到 Channel 关闭。
     * 关闭线程池并且释放所有的资源。
     */

    private EventLoopGroup eventLoopGroup;

    private Channel channel;

    private ClientHandler clientHandler;

    private String host;

    private Integer port;

    private static final int MAX_RETRY = 5;

    public NettyClient(String host, Integer port) {
        this.host = host;
        this.port = port;
    }

    public void connect() {
        clientHandler = new ClientHandler();
        eventLoopGroup = new NioEventLoopGroup();
        //启动类
        //创建Bootstrap
        Bootstrap bootstrap = new Bootstrap();
        //指定eventLoopGroup处理客户端事件
        bootstrap.group(eventLoopGroup)
                //指定传输使用的Channel
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                //在创建channel时，向channelPipeline中添加一个clientHandler实例
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(clientHandler);
                    }
                });
        // 链接服务器
        //设置服务器的地址、端口
        connect(bootstrap, host, port, MAX_RETRY);
//        ChannelFuture future = bootstrap.connect(host, port).sync();
//        channel = future.channel();
    }

    /**
     * 失败重连机制，参考Netty入门实战掘金小册
     *
     * @param bootstrap
     * @param host
     * @param port
     * @param retry
     */
    private void connect(Bootstrap bootstrap, String host, int port, int retry) {
        ChannelFuture channelFuture = bootstrap.connect(host, port).addListener(future -> {
            if (future.isSuccess()) {
                log.info("连接服务端成功");
            } else if (retry == 0) {
                log.error("重试次数已用完，放弃连接");
            } else {
                //第几次重连：
                int order = (MAX_RETRY - retry) + 1;
                //本次重连的间隔
                int delay = 1 << order;
                log.error("{} : 连接失败，第 {} 重连....", new Date(), order);
                bootstrap.config().group().schedule(() -> connect(bootstrap, host, port, retry - 1), delay, TimeUnit.SECONDS);
            }
        });
        try {
            boolean start = channelFuture.isDone();
            log.info("start",start);
            //连接到远程节点，阻塞，直到连接完成
            channel = channelFuture.sync().channel();
            boolean end = channelFuture.isDone();
            log.info("end",end);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送消息
     *
     * @param request
     * @return
     */
    public RpcResponse send(final RpcRequest request) {
        try {
//            channel.writeAndFlush(request).await();
            //数据冲刷
            ChannelFuture channelFuture = channel.writeAndFlush(request);
            //添加ChannelFutureListener以便在写操作完成后接收通知
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    //写操作完成，并没有错误发生
                    if (channelFuture.isSuccess()){
                        System.out.println("successful");
                    }else{
                        //记录错误
                        System.out.println("error");
                        channelFuture.cause().printStackTrace();
                    }
                }
            }).sync();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return clientHandler.getRpcResponse(request.getRequestId());
    }
    @PreDestroy
    public void close() {
        eventLoopGroup.shutdownGracefully();
        channel.closeFuture().syncUninterruptibly();
    }
}
